package databus.client.consumer;

import com.google.protobuf.InvalidProtocolBufferException;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import studio.raptor.databus.client.DatabusConfig;
import studio.raptor.databus.client.DatabusConsumer;
import studio.raptor.databus.client.DatabusConsumerBuilder;
import studio.raptor.databus.client.EventMessage;
import studio.raptor.databus.command.Event;

import javax.jms.JMSException;
import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Objects;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.locks.LockSupport;

/**
 * Created on 2017/9/4  <br>
 * databus应用下多个订阅者订阅同一主题下对应的死信队列
 *
 * @author koalton
 */
public class CustomerSubscriberTDLQ {

    DatabusConsumerBuilder consumerBuilder = null;
    public static final SimpleDateFormat Y2MS = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
    Logger log = LoggerFactory.getLogger(CustomerCenter.class);
    //两次获取消息的最大时间间隔-60s
    public static final int MAX_WAIT_RECEIVE_GAP_TIME = 60;
    //两次获取消息的初始化时间间隔-1s
    public static final int INIT_RECEIVE_GAP_TIME = 1;
    //两次获取消息时间间隔增长步长 2s
    public static final int RECEIVE_GAP_TIME_STEP = 2;

    public CustomerSubscriberTDLQ() {
        try {
            //实例配置对象
            DatabusConfig config = new DatabusConfig("Customer");
            //通过配置实例化builder对象
            consumerBuilder = new DatabusConsumerBuilder(config);
        } catch (JMSException e1) {
            log.info(e1.toString());
            e1.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
            log.info(e.getMessage());
            e.printStackTrace();
        }
    }

    /**
     * 多个订阅者订阅同一个事件
     */
    @Test
    public void subscribe() {
        //持久化订阅者客户端ID
        String[] clientId = {"111", "112"};
        //持久化订阅者名称
        String[] durableSubscriberName = {"ITV", "ODD"};
        int threads = 2;
        ExecutorService executorService = Executors.newFixedThreadPool(threads);
        for (int i = 0; i < threads; i++) {
            try {
                DatabusConsumer consumer = this.initConsumer(consumerBuilder, "AddOrder", 100001, null, clientId[i], durableSubscriberName[i]);
                String subName = durableSubscriberName[i];

                executorService.execute(new Runnable() {
                    DatabusConsumer consumer = null;
                    String durableSubscriberName  ;
                    public Runnable config(DatabusConsumer consumer, String durableSubscriberName){
                        this.consumer =  consumer;
                        this.durableSubscriberName = durableSubscriberName ;
                        return this;
                    }
                    @Override
                    public void run() {
                        int waitTime = INIT_RECEIVE_GAP_TIME;
                        while (true) {
                            EventMessage eventMessage = null;
                            try {
                                eventMessage = consumer.receiveUnACKMessage(1 * 1000);
                            } catch (JMSException e) {
                                e.printStackTrace();
                            }
                            if (eventMessage == null) {
                                System.out.println("At "
                                        + Y2MS.format(new Date())
                                        + " subscriber 【"+durableSubscriberName+"】 receive: empty");
                                LockSupport.parkUntil(System.currentTimeMillis() + waitTime * 1000);
                                waitTime += RECEIVE_GAP_TIME_STEP;
                                if (waitTime > MAX_WAIT_RECEIVE_GAP_TIME) {
                                    waitTime = MAX_WAIT_RECEIVE_GAP_TIME;
                                }
                                continue;
                            }
                            waitTime = 1;
                            try {
                                System.out.println("At "
                                        + Y2MS.format(new Date())
                                        + " subscriber "+durableSubscriberName+" receive:"
                                        + eventCoreInfo(eventMessage.getEvent()));
                                eventMessage.acknowledge();
                                LockSupport.parkUntil(System.currentTimeMillis() + 1* 1000);
                            } catch (JMSException e) {
                                e.printStackTrace();
                            } catch (InvalidProtocolBufferException e) {
                                e.printStackTrace();
                            }
                        }
                    }
                }.config(consumer,subName));
            } catch (JMSException e) {
                e.printStackTrace();
            }

        }
        //线程池不再接受新任务
        executorService.shutdown();
        //主程序保持运行
        while (true){}
    }

    /**
     * 多个订阅者订阅同一个事件，且订阅者携带过滤器
     */
    @Test
    public void subscribeWithSelector() {
        //持久化订阅者客户端ID
        String[] clientId = {"111", "112"};
        //持久化订阅者名称
        String[] durableSubscriberName = {"ITV", "ODD"};
        int threads = 2;
        ExecutorService executorService = Executors.newFixedThreadPool(threads);
        String consumerSelector = " name1 = 'name1' and name3 = 1 and name4 = 4 ";
        consumerSelector = " name4 = 4 ";
        for (int i = 0; i < threads; i++) {
            try {
                DatabusConsumer consumer = this.initConsumer(consumerBuilder, "AddOrder", 100001, consumerSelector, clientId[i], durableSubscriberName[i]);
                String subName = durableSubscriberName[i];

                executorService.execute(new Runnable() {
                    DatabusConsumer consumer = null;
                    String durableSubscriberName  ;
                    public Runnable config(DatabusConsumer consumer, String durableSubscriberName){
                        this.consumer =  consumer;
                        this.durableSubscriberName = durableSubscriberName ;
                        return this;
                    }
                    @Override
                    public void run() {
                        int waitTime = INIT_RECEIVE_GAP_TIME;
                        while (true) {
                            EventMessage eventMessage = null;
                            try {
                                eventMessage = consumer.receiveUnACKMessage(1 * 1000);
                            } catch (JMSException e) {
                                e.printStackTrace();
                            }
                            if (eventMessage == null) {
                                System.out.println("At " + Y2MS.format(new Date()) + " subscriber 【"+durableSubscriberName+"】 receive: empty");
                                LockSupport.parkUntil(System.currentTimeMillis() + waitTime * 1000);
                                waitTime += RECEIVE_GAP_TIME_STEP;
                                if (waitTime > MAX_WAIT_RECEIVE_GAP_TIME) {
                                    waitTime = MAX_WAIT_RECEIVE_GAP_TIME;
                                }
                                continue;
                            }
                            waitTime = 1;
                            try {
                                System.out.println("At "
                                        + Y2MS.format(new Date())
                                        + " subscriber "+durableSubscriberName+" receive:"
                                        + eventCoreInfo(eventMessage.getEvent()));
                                eventMessage.acknowledge();
                                LockSupport.parkUntil(System.currentTimeMillis() + 1* 1000);
                            } catch (JMSException e) {
                                e.printStackTrace();
                            } catch (InvalidProtocolBufferException e) {
                                e.printStackTrace();
                            }
                        }
                    }
                }.config(consumer,subName));
            } catch (JMSException e) {
                e.printStackTrace();
            }

        }
        //线程池不再接受新任务
        executorService.shutdown();
        //主程序保持运行
        while (true){}
    }


    public String eventCoreInfo(Event event) {
        //事件来源
        String eventSource = event.getEventBasicData().getSource();
        //事件序号
        String eventSeq = event.getEventBasicData().getKindId() + "";
        //事件名称
        String eventName = event.getEventBasicData().getName();
        //业务编号
        String businessNo = event.getEventMetaData().getBusinessNo() + "";
        //业务数据
        String centent = event.getEventBusinessData().getContent();
        return "{EventSource:" + eventSource + ",eventSeq:" + eventSeq + ",eventName:" + eventName + ",businessNo:" + businessNo + "}";
    }

    /**
     * 实例消费者
     *
     * @param consumerBuilder  消费者创建者
     * @param eventName        消费者所需消费的事件名称
     * @param eventKindId      消费者所需消费的事件kindId
     * @param consumerSelector 消费者所携带的消息过滤器，如果为NULL表示不会过滤消息
     * @return
     */
    private DatabusConsumer initConsumer(DatabusConsumerBuilder consumerBuilder,
                                         String eventName,
                                         Integer eventKindId,
                                         String consumerSelector,
                                         String clientId,
                                         String durableSubscriberName) throws JMSException {
        if (Objects.equals(null, consumerBuilder)) {
            return null;
        }
        DatabusConsumer consumer = null;
        if (Objects.equals(null, consumerSelector) || Objects.equals("", consumerSelector)) {
            //创建不携带过滤器的消费者
            if (!Objects.equals(null, clientId)
                    && !Objects.equals("", clientId)
                    && !Objects.equals(null, durableSubscriberName)
                    && !Objects.equals("", durableSubscriberName)) {
                //创建持久订阅者
                consumer = consumerBuilder.createDLQ(eventName, eventKindId, clientId, durableSubscriberName);
            } else {
                //创建普通消费者
                consumer = consumerBuilder.createDLQ(eventName, eventKindId);
            }
        } else {
            if (!Objects.equals(null, clientId)
                    && !Objects.equals("", clientId)
                    && !Objects.equals(null, durableSubscriberName)
                    && !Objects.equals("", durableSubscriberName)) {
                //创建持久订阅者
                consumer = consumerBuilder.createDLQ(eventName, eventKindId, consumerSelector, clientId, durableSubscriberName);
            } else {
                //创建普通消费者
                consumer = consumerBuilder.createDLQ(eventName, eventKindId, consumerSelector);
            }
        }
        return consumer;
    }

}
